package com.saac.imf.service;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Service;

@Service
public class MqService {

    static String ak = "QJOBLE0TRC2BN2DQ52S2";
    static String sk = "OiUmXUjvHbeLbCYC7wJnGkvV1TpwyWTJ5cjrHOAw";
    static String nameserver = "172.26.17.6:8100;172.26.17.171:8100";
    static String topic = "XIY-IMF-FLIGHT";
    static String consumerGroup = "KNX-IMF-FLIGHT";
    public String connMq() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup,
                new AclClientRPCHook(new SessionCredentials(ak, sk)),
                new AllocateMessageQueueAveragely(),
                true,
                null);
        consumer.setNamesrvAddr(nameserver);
        consumer.setUseTLS(true);
        return "success";
    }

    public static void main(String[] args) {
        producerMessage("11111111111111111");
//        try {
//            consumerMessage();
//        } catch (MQClientException e) {
//            throw new RuntimeException(e);
//        }

//        producerMessage("22222222222222222222");
//        new Thread(new Thread(() -> {
//            int i = 0;
//            while (true){
//                try {
//                    Thread.sleep(1000L);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
//                producerMessage(i ++ + "");
//            }
//        })).start();
    }

    static void producerMessage(String msg){
        System.out.println("111111111111111");
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
//        DefaultMQProducer producer = new DefaultMQProducer("TTTT",
//                new AclClientRPCHook(new SessionCredentials(ak, sk)),
//                true,
//                null);
        producer.setNamesrvAddr("192.168.31.118:9876");
        producer.setUseTLS(true);
        producer.setInstanceName("ProducerGroupName");
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        Message message = new Message(topic, msg.getBytes());
        try {
            producer.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        } catch (RemotingException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        } catch (MQBrokerException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    static void consumerMessage() throws MQClientException {
        System.out.println("22222222222222222");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup,
//                new AclClientRPCHook(new SessionCredentials(ak, sk)),
//                new AllocateMessageQueueAveragely(),
//                true,
//                null);
        consumer.setNamesrvAddr("192.168.31.118:9876");
        consumer.setUseTLS(true);
        consumer.subscribe("TopicTest", "*");
        // 设置Consumer第一次启动是从哪个位置开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        // 注册消息监听器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt message : msgs) {
                // 处理消息
                System.out.println(new String(message.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        // 启动消费者
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}
